package run.jingjing.websocket.model;

import io.netty.util.Timeout;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import run.jingjing.websocket.endpoint.EndpointManager;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * @author: jingjing
 * @date: 2021/6/24
 * @description:
 */
@Slf4j
public class RunningTopic {

    private LWsGroup lWsGroup;

    private LWsTopic lWsTopic;

    private List<Timeout> timeouts = new CopyOnWriteArrayList<>();

    private Long createTime;

    private MessageListener messageListener;

    private RedisMessageListenerContainer container;

    public LWsGroup getlWsGroup() {
        return lWsGroup;
    }

    public void setlWsGroup(LWsGroup lWsGroup) {
        this.lWsGroup = lWsGroup;
    }

    public LWsTopic getlWsTopic() {
        return lWsTopic;
    }

    public void setlWsTopic(LWsTopic lWsTopic) {
        this.lWsTopic = lWsTopic;
    }

    public List<Timeout> getTimeouts() {
        return timeouts;
    }

    public void setTimeouts(List<Timeout> timeouts) {
        this.timeouts = timeouts;
    }

    public RedisMessageListenerContainer getContainer() {
        return container;
    }

    public void setContainer(RedisMessageListenerContainer container) {
        this.container = container;
    }

    public MessageListener getMessageListener() {
        return messageListener;
    }

    public void setMessageListener(MessageListener messageListener) {
        this.messageListener = messageListener;
    }

    public RunningTopic() {
        this.createTime = System.currentTimeMillis();
    }

    public ChannelTopic channelTopic() {
        return new ChannelTopic(lWsGroup.getName().concat(":").concat(lWsTopic.getName()).concat(":").concat(createTime.toString()));
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        RunningTopic that = (RunningTopic) o;
        return Objects.equals(lWsGroup.getName(), that.lWsGroup.getName()) &&
                Objects.equals(lWsTopic.getName(), that.lWsTopic.getName());
    }

    @Override
    public int hashCode() {
        return Objects.hash(lWsGroup.getName(), lWsTopic.getName());
    }

    public void stop() {
        EndpointManager.removeRunningTopic(this);
        for (Timeout timeout : timeouts) {
            stopForTimeount(timeout);
        }
        container.removeMessageListener(messageListener, channelTopic());
    }

    private void stopForTimeount(Timeout timeout) {
        if (timeout == null) {
            return;
        }
        if (timeout.isCancelled()) {
            return;
        }
        try {
            long start = System.currentTimeMillis();
            // timeout may be expired by coincidence
            while (timeout.isExpired() && System.currentTimeMillis() - start < 10 * 60 * 1000L) {
                Thread.sleep(3000);
            }
        } catch (InterruptedException e) {
            log.error("wait to close timeout occurred error", e);
        } finally {
            timeout.cancel();
        }
    }

    public void addTimeouts(Timeout timeout) {
        timeouts.add(timeout);
    }

    public void removeTimeout(Timeout timeout) {
        timeouts.remove(timeout);
    }
}
